package rabbitmq

import (
	"fmt"
	"new_for_deepseek_rabbitmq/config"
	"new_for_deepseek_rabbitmq/logger"
	"sync"
	"time"

	"github.com/streadway/amqp"
)

var (
	rabbitmqClient *amqp.Connection
	mu             sync.RWMutex
)

// InitRabbitMQ 初始化 RabbitMQ 连接
func InitRabbitMQ() error {

	// 直接使用固定配置
	url := fmt.Sprintf("amqp://%s:%s@%s:%d/%s", config.DefaultConfig.Rabbitmq.User, config.DefaultConfig.Rabbitmq.Password, config.DefaultConfig.Rabbitmq.Host, config.DefaultConfig.Rabbitmq.Port, config.DefaultConfig.Rabbitmq.Vhost)

	var err error
	rabbitmqClient, err = amqp.Dial(url)
	if err != nil {
		logger.Error("连接 RabbitMQ 失败: " + err.Error())
		return fmt.Errorf("连接RabbitMQ失败: %v", err)
	}

	// 使用锁保护对 rabbitmqClient 的赋值
	mu.Lock()
	rabbitmqClient = rabbitmqClient
	mu.Unlock()

	// 检查队列配置
	if len(queues) == 0 {
		logger.Error("没有找到队列配置")
		return fmt.Errorf("没有找到队列配置")
	}

	// 初始化声明交换机和队列
	ch, err := GetChannel()
	if err != nil {
		logger.Error("获取通道失败: " + err.Error())
		return fmt.Errorf("获取通道失败: %v", err)
	}
	defer ch.Close()

	// 声明交换机和队列
	for _, queue := range queues {
		if err := DeclareExchangeAndQueue(ch, queue); err != nil {
			logger.Error("声明交换机和队列失败: " + err.Error())
			return fmt.Errorf("声明交换机和队列失败: %v", err)
		}
	}

	// 监听连接关闭
	go func() {
		notifyClose := rabbitmqClient.NotifyClose(make(chan *amqp.Error))
		err := <-notifyClose
		logger.Error("RabbitMQ 连接已关闭: %v", err)

		for {
			time.Sleep(time.Second * 5)
			if err := InitRabbitMQ(); err != nil {
				logger.Error("RabbitMQ 重连失败: %v", err)
				continue
			}
			logger.Info("RabbitMQ 重连成功")
			break
		}
	}()

	return nil
}

// GetChannel 获取一个新的 channel
func GetChannel() (*amqp.Channel, error) {

	// 使用临时变量保存 client 引用，避免长时间持有锁
	mu.RLock()
	client := rabbitmqClient
	mu.RUnlock()

	if client == nil || client.IsClosed() {
		logger.Error("RabbitMQ 连接未初始化或已关闭")
		return nil, fmt.Errorf("RabbitMQ 连接未初始化或已关闭")
	}

	ch, err := client.Channel()
	if err != nil {
		logger.Error("创建通道失败: " + err.Error())
		return nil, fmt.Errorf("创建通道失败: %v", err)
	}

	// 修改 QoS 设置，增加预取数量
	err = ch.Qos(
		config.DefaultConfig.Rabbitmq.PrefetchCount, // prefetchCount，改为10或更大的值，根据你的处理能力设置
		0,     // prefetchSize
		false, // global
	)
	if err != nil {
		ch.Close()
		logger.Error("设置 QoS 失败: " + err.Error())
		return nil, fmt.Errorf("设置 QoS 失败: %v", err)
	}

	return ch, nil
}

// Publish 发送消息
func Publish(exchange, routingKey string, body []byte) error {
	ch, err := GetChannel()
	if err != nil {
		return err
	}
	defer ch.Close()

	return ch.Publish(
		exchange,
		routingKey,
		false,
		false,
		amqp.Publishing{
			ContentType: "application/json",
			Body:        body,
		},
	)
}

// DeclareExchangeAndQueue 声明交换机和队列并绑定
func DeclareExchangeAndQueue(ch *amqp.Channel, config QueueConfig) error {
	// 声明交换机
	if err := ch.ExchangeDeclare(
		config.ExchangeName,
		config.Kind,
		true,  // durable
		false, // auto-deleted
		false, // internal
		false, // no-wait
		nil,   // arguments
	); err != nil {
		return fmt.Errorf("声明交换机 %s 失败: %v", config.ExchangeName, err)
	}

	// 声明队列
	_, err := ch.QueueDeclare(
		config.QueueName,
		true,  // durable
		false, // auto-deleted
		false, // exclusive
		false, // no-wait
		nil,   // arguments
	)
	if err != nil {
		return fmt.Errorf("声明队列 %s 失败: %v", config.QueueName, err)
	}

	// 绑定队列
	if err := ch.QueueBind(
		config.QueueName,
		config.RoutingKey,
		config.ExchangeName,
		false,
		nil,
	); err != nil {
		return fmt.Errorf("绑定队列 %s 失败: %v", config.QueueName, err)
	}

	return nil
}

// ReleaseChannel 释放通道
func ReleaseChannel(ch *amqp.Channel) {
	if ch != nil {
		ch.Close()
	}
}
